package heap;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import bufmgr.BufMgrException;
import bufmgr.BufferPoolExceededException;
import bufmgr.HashEntryNotFoundException;
import bufmgr.HashOperationException;
import bufmgr.InvalidFrameNumberException;
import bufmgr.PageNotReadException;
import bufmgr.PagePinnedException;
import bufmgr.PageUnpinnedException;
import bufmgr.ReplacerException;

import diskmgr.DiskMgrException;
import diskmgr.FileEntryNotFoundException;
import diskmgr.FileIOException;
import diskmgr.InvalidPageNumberException;
import diskmgr.InvalidRunSizeException;
import diskmgr.Page;

import global.GlobalConst;
import global.PageId;
import global.RID;
import global.SystemDefs;

public class Heapfile extends Object implements GlobalConst {
	private String name;
	private ArrayList<PageId> pages;
	private int noOfRecords;
	private HashSet<Integer> pids;//
	private boolean delete;
	private HFPage current;

	/**
	 * Initialize. A null name produces a temporary heapfile which will be
	 * deleted by the destructor. If the name already denotes a file, the file
	 * is opened; otherwise, a new empty file is created.
	 * 
	 * @param name file name
	 * @throws HFException
	 * @throws HFBufMgrException
	 * @throws HFDiskMgrException
	 * @throws ReplacerException
	 * @throws HashOperationException
	 * @throws PageUnpinnedException
	 * @throws InvalidFrameNumberException
	 * @throws PageNotReadException
	 * @throws BufferPoolExceededException
	 * @throws PagePinnedException
	 * @throws BufMgrException
	 * @throws HashEntryNotFoundException
	 * @throws IOException
	 */
	public Heapfile(String name) throws HFException, HFBufMgrException,
			HFDiskMgrException, ReplacerException, HashOperationException,
			PageUnpinnedException, InvalidFrameNumberException,
			PageNotReadException, BufferPoolExceededException,
			PagePinnedException, BufMgrException, HashEntryNotFoundException,
			IOException {
		this.name = name;
		if (name == null) {
			PageId firstPageId;
			Page page = new Page();
			try {
				firstPageId = SystemDefs.JavabaseBM.newPage(page, 1);
			} catch (Exception e) {
				throw new HFBufMgrException();
			}
			pages = new ArrayList<PageId>();
			pids = new HashSet<Integer>();
			pages.add(firstPageId);
			pids.add(firstPageId.pid);
			current = new HFPage(page);
			try {
				current.init(firstPageId, page);
			} catch (Exception e) {
				throw new HFException();
			}
			try {
				SystemDefs.JavabaseBM.unpinPage(firstPageId, true);
			} catch (Exception e) {
				throw new HFBufMgrException();
			}
		} else {
			PageId firstPageId;
			Page page = new Page();
			try {
				firstPageId = SystemDefs.JavabaseDB.get_file_entry(name);
			} catch (Exception e) {
				throw new HFDiskMgrException();
			}
			if (firstPageId == null) {
				try {
					firstPageId = SystemDefs.JavabaseBM.newPage(page, 1);
					SystemDefs.JavabaseDB.add_file_entry(name, firstPageId);
				} catch (Exception e) {
					throw new HFBufMgrException();
				}
				try {
					SystemDefs.JavabaseBM.unpinPage(firstPageId, true);
				} catch (Exception e) {
					throw new HFBufMgrException();
				}
				pages = new ArrayList<PageId>();
				pids = new HashSet<Integer>();
				pages.add(firstPageId);
				pids.add(firstPageId.pid);
				try {
					SystemDefs.JavabaseBM.pinPage(firstPageId, page, false);
				} catch (Exception e) {
					throw new HFBufMgrException();
				}
				pids.add(firstPageId.pid);
				noOfRecords = 0;
				current = new HFPage(page);
				try {
					current.init(firstPageId, page);
				} catch (Exception e) {
					throw new HFException();
				}
				try {
					SystemDefs.JavabaseBM.unpinPage(firstPageId, true);
				} catch (Exception e) {
					throw new HFBufMgrException();
				}

				return;
			}

			pages = new ArrayList<PageId>();
			pids = new HashSet<Integer>();
			addHFPages(firstPageId);
		}
	}

	private void addHFPages(PageId firstPageId) throws ReplacerException,
			HashOperationException, PageUnpinnedException,
			InvalidFrameNumberException, PageNotReadException,
			BufferPoolExceededException, PagePinnedException, BufMgrException,
			IOException, HashEntryNotFoundException {
		HFPage hfpage = new HFPage();
		current = new HFPage();
		SystemDefs.JavabaseBM.pinPage(firstPageId, hfpage, false);
		current.openHFpage(hfpage);
		pages.add(firstPageId);
		pids.add(firstPageId.pid);
		noOfRecords += noOfRecords(hfpage);
		SystemDefs.JavabaseBM.unpinPage(firstPageId, false);
		PageId currentPageId = current.getNextPage();
		while (currentPageId.pid != 0 && currentPageId.pid != -1) {
			HFPage nhf = new HFPage();
			SystemDefs.JavabaseBM.pinPage(currentPageId, nhf, false);
			pages.add(currentPageId);
			pids.add(currentPageId.pid);
			noOfRecords += noOfRecords(nhf);
			SystemDefs.JavabaseBM.unpinPage(currentPageId, false);
			currentPageId = nhf.getNextPage();
		}
	}

	private int noOfRecords(HFPage hf) throws IOException {
		RID nextRecord = hf.firstRecord();
		int counter = 0;
		while (nextRecord != null) {
			nextRecord = hf.nextRecord(nextRecord);
			counter++;
		}
		return counter;
	}

	/**
	 * Delete the file from the database.
	 * 
	 * @throws DiskMgrException
	 * @throws FileIOException
	 * @throws InvalidPageNumberException
	 * @throws InvalidRunSizeException
	 * @throws FileEntryNotFoundException
	 */
	public void deleteFile() throws InvalidSlotNumberException,
			FileAlreadyDeletedException, InvalidTupleSizeException,
			HFBufMgrException, HFDiskMgrException, IOException,
			InvalidRunSizeException, InvalidPageNumberException,
			FileIOException, DiskMgrException, FileEntryNotFoundException {
		if (delete)
			throw new FileAlreadyDeletedException();
		// free pages, full pages
		for (int i = 0; i < pages.size(); i++)
			try {
				SystemDefs.JavabaseDB.deallocate_page((pages.get(i)));
			} catch (Exception e) {
				throw new HFDiskMgrException();
			}
		try {
			SystemDefs.JavabaseDB.delete_file_entry(name);
		} catch (Exception e) {
			throw new HFDiskMgrException();
		}
		pages = null;
		pids = null;
		noOfRecords = 0;
		delete = true;
	}

	/**
	 * Delete record from file with given rid.
	 * @param rid
	 * @return true record deleted false:record not found 
	 * @throws InvalidSlotNumberException
	 * @throws InvalidTupleSizeException
	 * @throws HFException
	 * @throws HFBufMgrException
	 * @throws HFDiskMgrException
	 * @throws Exception
	 */
	public boolean deleteRecord(RID rid) throws InvalidSlotNumberException,
			InvalidTupleSizeException, HFException, HFBufMgrException,
			HFDiskMgrException, Exception {
		// check in hash set
		PageId pid = rid.pageNo;
		if (!pids.contains(pid.pid)) {
			throw new InvalidUpdateException();
		} else {// found in hash table
			Page page = new Page();
			try {
				SystemDefs.JavabaseBM.pinPage(pid, page, false);
			} catch (Exception e) {
				throw new HFBufMgrException();
			}
			HFPage hfpage = new HFPage();
			hfpage.openHFpage(page);

			hfpage.deleteRecord(rid);// record removed from hfpage
			SystemDefs.JavabaseBM.unpinPage(pid, true);

			noOfRecords--;
			return true;
		}
	}
	
	/**
	 * Return number of records in file. 
	 * @return
	 * @throws InvalidSlotNumberException
	 * @throws InvalidTupleSizeException
	 * @throws HFDiskMgrException
	 * @throws HFBufMgrException
	 * @throws IOException
	 */
	public int getRecCnt() throws InvalidSlotNumberException,
			InvalidTupleSizeException, HFDiskMgrException, HFBufMgrException,
			IOException {
		return noOfRecords;

	}
	
	/**
	 * Read record from file, returning pointer and length. 
	 * @param rid
	 * @return a Tuple. if Tuple==null, no more tuple
	 * @throws InvalidSlotNumberException
	 * @throws InvalidTupleSizeException
	 * @throws HFException
	 * @throws HFDiskMgrException
	 * @throws HFBufMgrException
	 * @throws Exception
	 */
	public Tuple getRecord(RID rid) throws InvalidSlotNumberException,
			InvalidTupleSizeException, HFException, HFDiskMgrException,
			HFBufMgrException, Exception {
		HFPage page = getPage(rid);
		return page.returnRecord(rid);
	}

	/**
	 * Insert record into file, return its Rid.
	 * 
	 * @param recPtr
	 * @return
	 * @throws InvalidSlotNumberException
	 * @throws InvalidTupleSizeException
	 * @throws SpaceNotAvailableException
	 * @throws HFException
	 * @throws HFBufMgrException
	 * @throws HFDiskMgrException
	 * @throws IOException
	 * @throws DiskMgrException
	 * @throws BufMgrException
	 * @throws PageNotReadException
	 * @throws PageUnpinnedException
	 * @throws PagePinnedException
	 * @throws InvalidFrameNumberException
	 * @throws HashEntryNotFoundException
	 * @throws ReplacerException
	 * @throws HashOperationException
	 * @throws BufferPoolExceededException
	 */
	public RID insertRecord(byte recPtr[]) throws InvalidSlotNumberException,
			InvalidTupleSizeException, SpaceNotAvailableException, HFException,
			HFBufMgrException, HFDiskMgrException, IOException,
			BufferPoolExceededException, HashOperationException,
			ReplacerException, HashEntryNotFoundException,
			InvalidFrameNumberException, PagePinnedException,
			PageUnpinnedException, PageNotReadException, BufMgrException,
			DiskMgrException {
		// Update the number of records
		if (recPtr.length > GlobalConst.MINIBASE_PAGESIZE)
			throw new SpaceNotAvailableException();
		noOfRecords++;
		for (int i = 0; i < pages.size(); i++) {
			PageId pid = pages.get(i);
			Page page = new Page();
			try {
				SystemDefs.JavabaseBM.pinPage(pid, page, false);
			} catch (Exception e) {
				throw new HFBufMgrException();
			}
			HFPage hfpage = new HFPage();
			hfpage.openHFpage(page);
			if (hfpage.available_space() >= recPtr.length) {
				RID rid = hfpage.insertRecord(recPtr);
				try {
					SystemDefs.JavabaseBM.unpinPage(pid, true);
				} catch (Exception e) {
					throw new HFBufMgrException();
				}
				return rid;
			}
			try {
				SystemDefs.JavabaseBM.unpinPage(pid, false);
			} catch (Exception e) {
				throw new HFBufMgrException();
			}
		}
		Page npage = new Page();
		PageId id = SystemDefs.JavabaseBM.newPage(npage, 1);
		HFPage nhfPage = new HFPage(npage);
		nhfPage.init(id, npage);
		RID rid = nhfPage.insertRecord(recPtr);
		pages.add(id);
		pids.add(id.pid);
		current.setNextPage(id);
		nhfPage.setPrevPage(current.getCurPage());
		current = nhfPage;
		SystemDefs.JavabaseBM.unpinPage(id, true);
		return rid;
	}

	/**
	 * 
	 * @returns an object of class scan
	 * @throws InvalidTupleSizeException
	 * @throws IOException
	 * @throws ReplacerException
	 * @throws HashOperationException
	 * @throws PageUnpinnedException
	 * @throws InvalidFrameNumberException
	 * @throws PageNotReadException
	 * @throws BufferPoolExceededException
	 * @throws PagePinnedException
	 * @throws BufMgrException
	 * @throws HashEntryNotFoundException
	 */
	public Scan openScan() throws InvalidTupleSizeException, IOException,
			ReplacerException, HashOperationException, PageUnpinnedException,
			InvalidFrameNumberException, PageNotReadException,
			BufferPoolExceededException, PagePinnedException, BufMgrException,
			HashEntryNotFoundException {
		return new Scan(this);
	}

	/**
	 * Updates the specified record in the heapfile.
	 * 
	 * @param rid
	 * @param newtuple
	 * @return
	 * @throws InvalidSlotNumberException
	 * @throws InvalidUpdateException
	 * @throws InvalidTupleSizeException
	 * @throws HFException
	 * @throws HFDiskMgrException
	 * @throws HFBufMgrException
	 * @throws Exception
	 */
	public boolean updateRecord(RID rid, Tuple newtuple)
			throws InvalidSlotNumberException, InvalidUpdateException,
			InvalidTupleSizeException, HFException, HFDiskMgrException,
			HFBufMgrException, Exception {
		HFPage hfpage = getPage(rid);
		Tuple oldTuple = hfpage.returnRecord(rid);
		if (newtuple.getLength() != oldTuple.getLength()) {
			throw new heap.InvalidUpdateException();
		}
		oldTuple.tupleCopy(newtuple);
		try {
			SystemDefs.JavabaseBM.unpinPage(rid.pageNo, false);
		} catch (Exception e) {
			throw new HFBufMgrException();
		}
		return true;
	}

	/**
	 * return the page contains the rid
	 * 
	 * @param rid
	 * @return
	 * @throws InvalidUpdateException
	 * @throws HFBufMgrException
	 */
	public HFPage getPage(RID rid) throws InvalidUpdateException,
			HFBufMgrException {
		PageId pid = rid.pageNo;
		if (!pids.contains(pid.pid)) {
			throw new InvalidUpdateException();
		}
		Page page = new Page();
		try {
			SystemDefs.JavabaseBM.pinPage(pid, page, false);
		} catch (Exception e) {
			throw new HFBufMgrException();
		}

		HFPage hfpage = new HFPage();
		hfpage.openHFpage(page);
		return hfpage;
	}

	/**
	 * 
	 * @return Iterator
	 */
	public Iterator<PageId> iterator() {
		return pages.iterator();
	}
}